package cn.py.boot_lua.asyn.updateAsyncMeth;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.function.Consumer;

public class AsyncQueueTaskEngine<T> {

    private static final Logger log = LoggerFactory.getLogger(AsyncQueueTaskEngine.class);

    private final BlockingQueue<T> taskQueue;
    private final Executor executor;
    private final Semaphore semaphore;
    private final Consumer<T> taskHandler;
    private final Consumer<Exception> errorHandler;
    private volatile boolean running = true;
    private final Thread consumerThread;
    private final String engineName;

    // 可配置：队列容量，默认 1000
    private static final int DEFAULT_QUEUE_CAPACITY = 100;
    private static final long SUBMIT_TIMEOUT_MS = 5000; // 提交超时时间

    public AsyncQueueTaskEngine(String engineName,
                                Executor executor,
                                int maxConcurrency,
                                Consumer<T> taskHandler) {
        this(engineName, executor, maxConcurrency, taskHandler, null, DEFAULT_QUEUE_CAPACITY);
    }

    public AsyncQueueTaskEngine(String engineName,
                                Executor executor,
                                int maxConcurrency,
                                Consumer<T> taskHandler,
                                Consumer<Exception> errorHandler,
                                int queueCapacity) {
        this.engineName = engineName;
        this.executor = executor;
        this.taskHandler = taskHandler;
        this.errorHandler = errorHandler != null ? errorHandler : this::defaultErrorHandler;
        this.taskQueue = new LinkedBlockingQueue<>(queueCapacity);
        this.semaphore = maxConcurrency > 0 ? new Semaphore(maxConcurrency) : null;

        this.consumerThread = new Thread(this::consumeLoop, "AsyncQueue-" + engineName + "-Consumer");
        this.consumerThread.setDaemon(true);
        this.consumerThread.start();

        log.info("AsyncQueueTaskEngine [{}] 启动，maxConcurrency={}, queueCapacity={}",
                engineName, maxConcurrency, queueCapacity);
    }

    /**
     * 提交任务（带超时）
     */
    public boolean submitTask(T task) {
        if (task == null) return false;

        try {
            if (taskQueue.offer(task, SUBMIT_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
                log.debug("[{}] 任务入队成功: {}", engineName, task);
                return true;
            } else {
                log.warn("[{}] 任务提交超时或队列已满，丢弃任务: {}", engineName, task);
                return false;
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.warn("[{}] 提交任务被中断: {}", engineName, task);
            return false;
        }
    }

    /**
     * 消费循环
     */
    private void consumeLoop() {
        log.debug("[{}] 消费者线程启动", engineName);
        while (running) {
            try {
                log.trace("[{}] 等待新任务...", engineName);
                T task = taskQueue.take();
                log.debug("[{}] 获取任务: {}", engineName, task);

                Runnable runnable = () -> {
                    boolean acquired = false;
                    try {
                        // 在 executor 线程中获取信号量
                        if (semaphore != null) {
                            try {
                                semaphore.acquire();
                                acquired = true;
                            } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                                log.warn("[{}] 任务执行前被中断（信号量）: {}", engineName, task);
                                return;
                            }
                        }

                        log.info("[{}] 开始执行任务: {}", engineName, task);
                        taskHandler.accept(task);
                        log.info("[{}] 任务执行完成: {}", engineName, task);

                    } catch (Exception e) {
                        log.error("[{}] 任务执行失败: {}", engineName, task, e);
                        errorHandler.accept(e);
                    } finally {
                        if (acquired && semaphore != null) {
                            semaphore.release();
                            log.trace("[{}] 释放信号量，当前可用: {}", engineName, semaphore.availablePermits());
                        }
                    }
                };

                executor.execute(runnable);

            } catch (InterruptedException e) {
                if (!running) {
                    log.info("[{}] 消费者线程收到中断，准备退出", engineName);
                    Thread.currentThread().interrupt();
                    break;
                } else {
                    log.warn("[{}] 消费循环被意外中断", engineName, e);
                }
            } catch (Exception e) {
                log.error("[{}] 消费循环发生未预期异常", engineName, e);
            }
        }
        log.info("[{}] 消费者线程已退出", engineName);
    }

    /**
     * 默认异常处理器
     */
    private void defaultErrorHandler(Exception e) {
        // 可集成告警、上报等
        log.info("错误异常: {}", e.getMessage());
    }

    /**
     * 优雅关闭
     */
    public void shutdown() {
        log.info("[{}] 正在关闭引擎...", engineName);
        running = false;
        consumerThread.interrupt();

        try {
            consumerThread.join(5000); // 最多等待 5 秒
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.warn("[{}] 关闭时等待线程中断", engineName);
        }

        // 清空队列 (根据自己的业务决定)
        int remaining = taskQueue.size();
        if (remaining > 0) {
            log.warn("[{}] 关闭后仍有 {} 个任务未处理", engineName, remaining);
        }

        log.info("[{}] 已关闭", engineName);
    }

    // --- Getter for monitoring ---
    public int getQueueSize() {
        return taskQueue.size();
    }

    public int getRemainingCapacity() {
        return taskQueue.remainingCapacity();
    }

    public int getActivePermits() {
        return semaphore == null ? 0 : semaphore.availablePermits();
    }

    public boolean isRunning() {
        return running;
    }

}
